package com.stan.core.spark.ad.realTimeFlowCount;

import com.stan.core.mapper.AdRealTimeCountMapper;
import com.stan.core.mapper.factory.MapperFactory;
import com.stan.core.vo.AdRealTimeCount;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;


/**
 * 功能1 流量实时统计结果
 * 目标数据 : (yyyyMMdd_province_city_aid,clickCount)
 */
public class ComputeRealTimeCountFunction
        implements Function<JavaPairRDD<String, String>, JavaPairRDD<String, Integer>> {
    /**
     * 算法思路:
     *  1.进行字段顺序的调整，调整为(yyyyMMdd_province_city_aid,1)
     *  2.按key进行聚合计数 ， 获得(yyyyMMdd_province_city_aid,currentClickCount)
     *  3.与数据库中的数据进行累加 (yyyyMMdd_province_city_aid,clickCount) 并更新到数据库
     *  输入为 (userid,data)
     *  例如: (20,2019-04-20 10:09:42 上海 上海 20 33)
     */
    @Override
    public JavaPairRDD<String, Integer> call(
            JavaPairRDD<String, String> stringStringJavaPairRDD) throws Exception {
        // 1.进行字段顺序的调整
        // (20,2019-04-20 10:09:42 上海 上海 20 33)
        JavaPairRDD<String, Integer> dateProvinceCityAid2OneRDD = stringStringJavaPairRDD.mapToPair(
                new PairFunction<Tuple2<String, String>, String, Integer>() {
                    @Override
                    public Tuple2<String, Integer> call(Tuple2<String, String> stringStringTuple2) throws Exception {
                        String data = stringStringTuple2._2;
                        String[] dataSplited = data.split(" ");
                        String date = dataSplited[0];
                        String province = dataSplited[2];
                        String city = dataSplited[3];
                        String aid = dataSplited[5];
                        String newKey = date + "_" + province + "_" + city + "_" + aid;
                        return new Tuple2<String,Integer>(newKey,1);
                    }
                }
        );

        // 2.按key进行聚合计数 ， 获得(yyyyMMdd_province_city_aid,currentClickCount)
        JavaPairRDD<String,Integer> dateProvinceCityAid2CurrentCountRDD = dateProvinceCityAid2OneRDD.reduceByKey(
                new Function2<Integer, Integer, Integer>() {
                    @Override
                    public Integer call(Integer integer, Integer integer2) throws Exception {
                        return integer + integer2;
                    }
                }
        );

        // 3.与数据库中的数据进行累加 (yyyyMMdd_province_city_aid,clickCount)
        JavaPairRDD<String,Integer> dateProvinceCityAid2CountRDD = dateProvinceCityAid2CurrentCountRDD.mapToPair(
                new PairFunction<Tuple2<String, Integer>, String, Integer>() {
                    @Override
                    public Tuple2<String, Integer> call(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
                        String dateProvinceCityAid = stringIntegerTuple2._1;
                        String[] strSplited = dateProvinceCityAid.split("_");
                        String date = strSplited[0];
                        String province =strSplited[1];
                        String city = strSplited[2];
                        String aid = strSplited[3];

                        int currentClickCount = stringIntegerTuple2._2;

                        AdRealTimeCountMapper adRealTimeCountMapper = MapperFactory.
                                getMapperFactory().getMapper(AdRealTimeCountMapper.class);


                        AdRealTimeCount adRealTimeCount = new AdRealTimeCount(date,province,city,aid);

                        adRealTimeCount = adRealTimeCountMapper.
                                getAdRealTimeCount(adRealTimeCount);
                        System.out.println(dateProvinceCityAid + adRealTimeCount);
                        if(adRealTimeCount == null){
                            adRealTimeCount = new AdRealTimeCount(date,province,city,aid,currentClickCount);
                            adRealTimeCountMapper.addAdRealTimeCount(adRealTimeCount);
                        }else{
                            adRealTimeCount.setClickCount(adRealTimeCount.getClickCount() + currentClickCount);
                            adRealTimeCount.setClickCount(adRealTimeCount.getClickCount());
                            adRealTimeCount.setClickCount(adRealTimeCount.getClickCount());
                            adRealTimeCountMapper.updateAdRealTimeCount(adRealTimeCount);
                        }
                        return new Tuple2<String, Integer>(dateProvinceCityAid,adRealTimeCount.getClickCount());
                    }
                }
        );
        return dateProvinceCityAid2CountRDD;
    }
}
